package com.atguigu.gamll.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.app.dwd.BaseApp;
import com.atguigu.gamll.realtime.beans.VisitorStats;
import com.atguigu.gamll.realtime.utils.ClickHouseUtil;
import com.atguigu.gamll.realtime.utils.DateUtil;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

// 访客主题统计
public class VisitorStatsApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        VisitorStatsApp visitorStatsApp = new VisitorStatsApp();
        visitorStatsApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 3.从kafka主题读取数据
        // 3.1 声明消费的主题以及消费者组
        String pageLogTopic = "dwd_page_log";
        String uvTopic = "dwm_unique_visitor";
        String ujdTopic = "dwm_user_jump_detail";
        String groupId = "visitor_stats_group";
        // 3.2 创建消费者对象
        FlinkKafkaConsumer<String> pageViewSource = MyKafkaUtil.getKafkaSource(pageLogTopic, groupId);
        FlinkKafkaConsumer<String> uniqueVisitSource = MyKafkaUtil.getKafkaSource(uvTopic, groupId);
        FlinkKafkaConsumer<String> userJumpSource = MyKafkaUtil.getKafkaSource(ujdTopic, groupId);
        // 3.3 消费数据，封装成流
        DataStreamSource<String> pageLogStrDS = env.addSource(pageViewSource);
        DataStreamSource<String> uvStrDS = env.addSource(uniqueVisitSource);
        DataStreamSource<String> ujdStrDS = env.addSource(userJumpSource);

        // TODO 4.对不同流的数据进行类型转换 jsonStr -> VisitorStats
        // 页面日志
        SingleOutputStreamOperator<VisitorStats> pageLogStatsDS = pageLogStrDS.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String jsonStr) throws Exception {
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                VisitorStats visitorStats = new VisitorStats(
                        "",
                        "",
                        commonJsonObj.getString("vc"),
                        commonJsonObj.getString("ch"),
                        commonJsonObj.getString("ar"),
                        commonJsonObj.getString("is_new"),
                        0L, 1L, 0L, 0L,
                        pageJsonObj.getLong("during_time"),
                        jsonObj.getLong("ts")
                );
                String lastPageId = pageJsonObj.getString("last_page_id");
                if (lastPageId == null || lastPageId.length() == 0) {
                    visitorStats.setSv_ct(1L);
                }
                return visitorStats;
            }
        });
        // 独立访客
        SingleOutputStreamOperator<VisitorStats> uvStatsDS = uvStrDS.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String jsonStr) throws Exception {
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                return new VisitorStats(
                        "",
                        "",
                        commonJsonObj.getString("vc"),
                        commonJsonObj.getString("ch"),
                        commonJsonObj.getString("ar"),
                        commonJsonObj.getString("is_new"),
                        1L, 0L, 0L, 0L, 0L, jsonObj.getLong("ts")
                );
            }
        });
        // 用户跳出
        SingleOutputStreamOperator<VisitorStats> ujdStatsDS = ujdStrDS.map(
                new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        return new VisitorStats(
                                "",
                                "",
                                commonJsonObj.getString("vc"),
                                commonJsonObj.getString("ch"),
                                commonJsonObj.getString("ar"),
                                commonJsonObj.getString("is_new"),
                                0L, 0L, 0L, 1L, 0L, jsonObj.getLong("ts")
                        );
                    }
                }
        );

        // TODO 5.使用union进行合并
        DataStream<VisitorStats> unionDS = pageLogStatsDS.union(uvStatsDS, ujdStatsDS);

        // TODO 6.设置水位线
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWatermarkDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                    @Override
                    public long extractTimestamp(VisitorStats visitorStats, long l) {
                        return visitorStats.getTs();
                    }
                })
        );

        // TODO 7.分组 注意：不要按照mid进行分组，因为单位窗口内，不会有太多数据产生，起不到聚合效果
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> keyedDS = visitorStatsWithWatermarkDS.keyBy(
                new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                        return Tuple4.of(
                                visitorStats.getVc(),
                                visitorStats.getCh(),
                                visitorStats.getAr(),
                                visitorStats.getIs_new()
                        );
                    }
                }
        );

        // TODO 8.开窗
        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // TODO 9.聚合计算
        SingleOutputStreamOperator<VisitorStats> reduceDS = windowDS.reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats stats1, VisitorStats stats2) throws Exception {
                        //把度量数据两两相加
                        stats1.setPv_ct(stats1.getPv_ct() + stats2.getPv_ct());
                        stats1.setUv_ct(stats1.getUv_ct() + stats2.getUv_ct());
                        stats1.setUj_ct(stats1.getUj_ct() + stats2.getUj_ct());
                        stats1.setSv_ct(stats1.getSv_ct() + stats2.getSv_ct());
                        stats1.setDur_sum(stats1.getDur_sum() + stats2.getDur_sum());
                        return stats1;
                    }
                },
                new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<VisitorStats> input, Collector<VisitorStats> out) throws Exception {
                        for (VisitorStats visitorStats : input) {
                            visitorStats.setStt(DateUtil.toYMDHMS(new Date(window.getStart())));
                            visitorStats.setEdt(DateUtil.toYMDHMS(new Date(window.getEnd())));
                            visitorStats.setTs(System.currentTimeMillis());
                            out.collect(visitorStats);
                        }
                    }
                }
        );
        reduceDS.print(">>>");

        // TODO 10.将聚合计算的结果写到ClickHouse数据库
        reduceDS.addSink(
                ClickHouseUtil.getSinkFunction("insert into visitor_stats_0819 values(?,?,?,?,?,?,?,?,?,?,?,?)")
        );
    }
}
